package com.liujt.util.redis;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.util.ByteArrayWrapper;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.scheduling.SchedulingAwareRunnable;

/**
 * Created by liujt on 2017/2/21.
 */
public class MessageListenerContainer implements DisposableBean {

    private Logger logger = LoggerFactory.getLogger(MessageListenerContainer.class);

    private final Map<ByteArrayWrapper, MessageListener> listenerMap = new ConcurrentHashMap<>();

    private ExecutorService subscriptionExecutor;

    private ExecutorService taskExecutor;

    private RedisConnectionFactory connectionFactory;

    private volatile boolean listening = false;

    private volatile boolean running = false;

    private final SubscriptionTask subscriptionTask = new SubscriptionTask();

    private volatile RedisSerializer<String> keySerializer = new StringRedisSerializer();

    public void setConnectionFactory(RedisConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    public void addMessageListener(String key, MessageListener listener) {
        ByteArrayWrapper holder = new ByteArrayWrapper(keySerializer.serialize(key));
        listenerMap.put(holder, listener);
        lazyListen();
    }

    protected void processMessage(MessageListener listener, Message message) {
        try {
            listener.onMessage(message);
        } catch (Throwable e) {
            logger.error("Execution of message listener failed, and no ErrorHandler has been set.", e);
        }
    }

    private void lazyListen() {
        if (running) {
            if (!listening) {
                synchronized (this) {
                    if (!listening) {
                        if (listenerMap.size() > 0) {
                            subscriptionExecutor.execute(subscriptionTask);
                            listening = true;
                        }
                    }
                }
            }
        }
    }

    @Override public void destroy() throws Exception {
        running = false;
        subscriptionTask.cancel();
        subscriptionExecutor.shutdown();
        taskExecutor.shutdown();
    }

    @PostConstruct
    public void afterPropertiesSet() throws Exception {
        if (!running) {
            running = true;
            subscriptionExecutor = Executors.newSingleThreadExecutor();
            taskExecutor = new ThreadPoolExecutor(5, 50, 1, TimeUnit.MINUTES,
                new LinkedBlockingQueue<Runnable>());
            lazyListen();
        }
    }

    private class SubscriptionTask implements SchedulingAwareRunnable {

        private volatile boolean stop = false;
        private volatile RedisConnection connection;

        @Override public boolean isLongLived() {
            return true;
        }

        @Override public void run() {
            try {
                connection = connectionFactory.getConnection();
                while (!stop) {
                    dispatchMessage();
                    Thread.sleep(100);
                }
            } catch (Exception e) {
                logger.error("", e);
            }
        }

        void dispatchMessage() {
            for (final Map.Entry<ByteArrayWrapper, MessageListener> entry : listenerMap.entrySet()) {
                final byte[] key = entry.getKey().getArray();
                Long size = connection.lLen(key);
                if (size == null || size.longValue() == 0) continue;
                for (long i = 0; i < size; i++) {
                    final byte[] val = connection.rPop(key);
                    if (val == null) continue;
                    taskExecutor.submit(new Runnable() {
                        public void run() {
                            processMessage(entry.getValue(), new Message(keySerializer.deserialize(key), val));
                        }
                    });
                }
            }
        }

        void cancel() {
            listening = false;
            stop = true;
            closeConnection();
        }

        void closeConnection() {
            if (connection != null) {
                logger.trace("Closing connection");
                try {
                    connection.close();
                } catch (Exception e) {
                    logger.warn("Error closing subscription connection", e);
                }
                connection = null;
            }
        }
    }

}
